package com.flying.learning.si.hub;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.NewTopic;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
import org.springframework.integration.IntegrationMessageHeaderAccessor;
import org.springframework.integration.aggregator.BarrierMessageHandler;
import org.springframework.integration.annotation.Gateway;
import org.springframework.integration.annotation.MessagingGateway;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.kafka.dsl.Kafka;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.KafkaAdmin;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.messaging.Message;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.web.bind.annotation.*;

import java.util.HashMap;
import java.util.Map;

@SpringBootApplication
@RestController
@Slf4j
public class So483499931Application {
    private static final String TRANSACTION_TOPIC1 = "TRANSACTION_TOPIC3";
    private static final String TRANSACTION_RESULT1 = "TRANSACTION_RESULT3";

    public static void main(String[] args) {
        SpringApplication.run(So483499931Application.class, args);
    }

    private final Exchanger exchanger;
    private final KafkaTemplate<String, String> kafkaTemplate;

    @Autowired
    public So483499931Application(Exchanger exchanger, KafkaTemplate<String, String> kafkaTemplate) {
        this.exchanger = exchanger;
        this.kafkaTemplate = kafkaTemplate;
        kafkaTemplate.setDefaultTopic(TRANSACTION_RESULT1);
    }

    @RequestMapping(path = "/foo/{id}/{other}", method = RequestMethod.GET)
    @ResponseBody
    public String foo(@PathVariable String id, @PathVariable String other) {
        log.info("Controller received: " + other);
        String reply = this.exchanger.exchange(id, other);
        log.info("Controller replying: " + reply);
        return reply;
    }
    // Client side

    @MessagingGateway(defaultRequestChannel = "outbound", defaultReplyTimeout = "10000")
    public interface Exchanger {
        @Gateway
        String exchange(@Header(IntegrationMessageHeaderAccessor.CORRELATION_ID) String id, @Payload String out);
    }

    @Bean
    public IntegrationFlow router() {
        return IntegrationFlows.from("outbound")
                .routeToRecipients(r -> r
                        .recipient("toKafka")
                        .recipient("barrierChannel"))
                .get();
    }

    @Bean
    public IntegrationFlow outFlow(KafkaTemplate<String, String> kafkaTemplate) {
        return IntegrationFlows.from("toKafka")
                .handle(Kafka.outboundChannelAdapter(kafkaTemplate).topic(TRANSACTION_TOPIC1))
                .get();
    }

    @Bean
    public IntegrationFlow barrierFlow(BarrierMessageHandler barrier) {
        return IntegrationFlows.from("barrierChannel")
                .handle(barrier)
                .transform("payload.get(1)") // payload is list with input/reply
                .get();
    }

    @Bean
    public BarrierMessageHandler barrier() {
        return new BarrierMessageHandler(10_000L);
    }

    @KafkaListener(id = "clientReply", topics = TRANSACTION_RESULT1)
    public void result(Message<?> reply) {
        log.info("Received reply: " + reply.getPayload() + " for id "
                + reply.getHeaders().get(IntegrationMessageHeaderAccessor.CORRELATION_ID));
        barrier().trigger(reply);
    }

    // Server side

    @Bean
    public IntegrationFlow server(ConsumerFactory<String, String> consumerFactory, KafkaTemplate<String, String> kafkaTemplate) {
        return IntegrationFlows.from(Kafka.messageDrivenChannelAdapter(consumerFactory, TRANSACTION_TOPIC1))
                .handle("so483499931Application", "service")
                .handle(Kafka.outboundChannelAdapter(kafkaTemplate).topic(TRANSACTION_RESULT1))
                .get();
    }

    public String service(String in) throws InterruptedException {
        log.info("Service Received " + in);
        Thread.sleep(5_000);
        log.info("Service Replying to " + in);
        return "reply for " + in;
    }

    // Provision topics if needed

    // provided by Boot in 2.0
    @Bean
    public KafkaAdmin admin() {
        Map<String, Object> config = new HashMap<>();
        config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.6.249:9092");
        return new KafkaAdmin(config);
    }

    @Bean
    public NewTopic topic1() {
        return new NewTopic(TRANSACTION_TOPIC1, 10, (short) 1);
    }

    @Bean
    public NewTopic result1() {
        return new NewTopic(TRANSACTION_RESULT1, 10, (short) 1);
    }

}
